package ru.mipt.victator.pagerank.rank;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

public class RankReduce extends MapReduceBase implements Reducer
{
	float d;
	float StartPR;
	float k;
	
	@Override
	public void configure(JobConf job)
	{
		d = Float.parseFloat(job.get("RankJob.CalculationsParameter.Float"));
		StartPR = Float.parseFloat(job.get("InputData.StartPageRank.Float"));
		k = StartPR*(1-d);
	}

	@SuppressWarnings("unchecked")
	@Override
	public void reduce(WritableComparable key, Iterator values,
			OutputCollector output, Reporter reporter) throws IOException
	{
		float sum = k;
	    while (values.hasNext())
	    {
	    	sum += ((FloatWritable) values.next()).get();
	    }
	    output.collect(key, new FloatWritable(sum));
	}

}
